package com.example.literj.kt.internal.operators.single

import com.example.literj.kt.core.Scheduler
import com.example.literj.kt.core.Single
import com.example.literj.kt.core.SingleObserver
import com.example.literj.kt.core.SingleSource
import com.example.literj.kt.disposables.Disposable
import com.example.literj.kt.internal.disposables.DisposableHelper
import java.util.concurrent.atomic.AtomicReference

class SingleObserveOn<T>(
    private val source: SingleSource<out T>,
    private val scheduler: Scheduler
) : Single<T>() {
    override fun subscribeActual(observer: SingleObserver<in T>) {
        source.subscribe(ObserveOnSingleObserver(observer, scheduler))
    }

    internal class ObserveOnSingleObserver<T>(
        private val downstream: SingleObserver<in T>,
        private val scheduler: Scheduler
    ) :
        AtomicReference<Disposable>(), SingleObserver<T>, Disposable,
        Runnable {
        var value: T? = null
        var error: Throwable? = null

        override fun onSubscribe(d: Disposable) {
            if (DisposableHelper.setOnce(this, d)) {
                downstream.onSubscribe(this)
            }
        }

        override fun onSuccess(t: T) {
            this.value = t
            val d = scheduler.scheduleDirect(this)
            DisposableHelper.replace(this, d)
        }

        override fun onError(e: Throwable) {
            error = e
            val d = scheduler.scheduleDirect(this)
            DisposableHelper.replace(this, d)
        }

        override fun run() {
            val ex = error
            if (ex != null) {
                downstream.onError(ex)
            } else {
                downstream.onSuccess(value!!)
            }
        }

        override fun dispose() {
            DisposableHelper.dispose(this)
        }

        override fun isDisposed() = DisposableHelper.isDisposed(get())
    }
}